package com.ideal.frame.node.netty;

import cn.hutool.core.util.ObjectUtil;
import com.ideal.frame.node.entity.HeartbeatNodeTb;
import com.ideal.frame.node.entity.HeartbeatRootNodeTb;
import com.ideal.frame.node.entity.NodeTb;
import com.ideal.frame.node.entity.RootNodeTb;
import com.ideal.frame.node.enumeration.NodeLevel;
import com.ideal.frame.node.enumeration.NodeStatus;
import com.ideal.frame.node.netty.client.NettyClientConstant;
import com.ideal.frame.node.netty.client.NettyClientHandleInitializer;
import com.ideal.frame.node.service.HeartbeatRootNodeService;
import com.ideal.frame.node.service.HeartbeatService;
import com.ideal.frame.node.service.NodeService;
import com.ideal.frame.node.service.RootNodeService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Triple;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @description:
 * @author: xuyang
 * @date: 2021/9/2 9:33
 */
@Component
@Slf4j
//@ConditionalOnExpression("#{nodeService.canBeNettyClient()}")
public class NettyClient {

    private Bootstrap bootstrap;
    EventLoopGroup eventLoopGroup = new NioEventLoopGroup();

    private List<Triple<String, Integer, Long>> serverHostAndPortList = new ArrayList<>();
    private List<Triple<String, Integer, Long>> rootServerHostAndPortList = new ArrayList<>();

    @Value("${netty.client.reconn.delay:10}")
    private int reconn_delay;

    @Autowired
    NodeService nodeService;

    @Autowired
    RootNodeService rootNodeService;

    @Autowired
    HeartbeatService heartbeatService;

    @Autowired
    HeartbeatRootNodeService heartbeatRootNodeService;

    @Autowired
    NettyClientHandleInitializer handleInitializer;

    @PostConstruct
    public void initialize(){
        NodeTb node = nodeService.getByLevel();
        List<RootNodeTb> rootNodeList = rootNodeService.getAllNodes();
        rootServerHostAndPortList = rootNodeList.stream()
                .filter(item -> item.getNodeip() != null && item.getNettyport() != null)
                .map(item -> Triple.of(item.getNodeip(), item.getNettyport(), item.getId()))
                .collect(Collectors.toList());
        List<NodeTb> nodeList = nodeService.getAllNodes();
        serverHostAndPortList = nodeList.stream()
                .filter(item -> !item.getNodelevel().equals(node.getNodelevel()))
                .filter(item -> item.getNodeip() != null && item.getNettyport() != null)
                .map(item -> Triple.of(item.getNodeip(), item.getNettyport(), item.getId()))
                .collect(Collectors.toList());
        NettyClientConstant.selfNodeLevel = node.getNodelevel();
        NettyClientConstant.heartbeatMsg = new NettyMsg(node.getId(), node.getNodename(), "ping");
        // 初始化bootstrap
        bootstrap = new Bootstrap()
                .group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(handleInitializer);
        // 连接多个服务端
        for(Triple<String, Integer, Long> triple : serverHostAndPortList){
            this.doConncet(triple.getLeft(), triple.getMiddle(), triple.getRight(), node.getNodelevel());
        }
        // 连接多个服务端
        for(Triple<String, Integer, Long> triple : rootServerHostAndPortList){
            this.doConncet(triple.getLeft(), triple.getMiddle(), triple.getRight(), NodeLevel.CHILD_ROOT.getValue());
        }
    }

    /**
     * 连接服务端
     */
    public void doConncet(String host, Integer nettyPort, Long nodeId, Integer nodeLevel) {
        Channel channel = null;
        if(NettyClientConstant.channelMaps.get(host + ":" + nettyPort) != null){
            channel = NettyClientConstant.channelMaps.get(host + ":" + nettyPort).getChannel();
        }
        // 启动客户端
        if (channel != null && channel.isActive()) {
            return;
        }
        // 定时任务每分钟查询取消心跳检测的记录，取消心跳检测就从channelMaps中删除对应的记录
        String hostAndPort = host + ":" + nettyPort;
        if(NettyClientConstant.cancelHeartbeatList.contains(hostAndPort)){
            log.info("客户端已经取消与[{}]的心跳检测后，不能自动重连", hostAndPort);
            return;
        }
        // 更新自动重连的次数
        NettyClientConstant.incrRetryConnTimes(hostAndPort);
        // 开始自动重连
        ChannelFuture channelFuture = bootstrap.connect(host, nettyPort);
        channelFuture.addListener((ChannelFutureListener) futureListener -> {
            if (channelFuture.isSuccess()) {
                Channel channel2 = futureListener.channel();
                NettyClientConstant.channelMaps.put(host + ":" + nettyPort, new NettyClientConstant.ChannelAndNodeId(nodeId, nodeLevel,channel2));
                log.info("connect server[{}:{}] successfully", host, nettyPort);
                // 连接成功后，清除连接次数
                NettyClientConstant.retryConnTimesMap.remove(hostAndPort);
            } else {
                if (ObjectUtil.equal(NodeLevel.CHILD_ROOT.getValue(), nodeLevel)) {
                    this.saveOffline(nodeId);
                } else {
                    this.saveOfflineRoot(nodeId);
                }
                log.info("Failed to connect to server[{}:{}], try connect after {}s", host, nettyPort, reconn_delay);
                futureListener.channel().eventLoop().schedule(() -> doConncet(host, nettyPort, nodeId, nodeLevel), reconn_delay, TimeUnit.SECONDS);
            }
        });
    }

    /**
     * 连接失败时，将心跳日志入库
     */
    @Async
    public void saveOffline(Long nodeId){
        // 心跳日志入库
        HeartbeatNodeTb heartbeat = new HeartbeatNodeTb();
        heartbeat.setNodeid(nodeId.toString());
        heartbeat.setInsertdate(new Date());
        heartbeat.setUpdatedate(new Date());
        heartbeat.setStatus(NodeStatus.InActive.getValue());
        heartbeatService.save(heartbeat);
        // 更新nodeTb的status字段（反映节点的实时状态）
        NodeTb nodeTb = nodeService.getById(nodeId);
        if(nodeTb != null){
            nodeTb.setStatus(NodeStatus.InActive.getBoolValue());
            nodeService.save(nodeTb);
        }
    }

    /**
     * 连接失败时，将心跳日志入库
     */
    @Async
    public void saveOfflineRoot(Long nodeId){
        // 心跳日志入库
        HeartbeatRootNodeTb heartbeat = new HeartbeatRootNodeTb();
        heartbeat.setNodeid(nodeId.toString());
        heartbeat.setInsertdate(new Date());
        heartbeat.setUpdatedate(new Date());
        heartbeat.setStatus(NodeStatus.InActive.getValue());
        heartbeatRootNodeService.save(heartbeat);
        // 更新nodeTb的status字段（反映节点的实时状态）
        RootNodeTb nodeTb = rootNodeService.getById(nodeId);
        if(nodeTb != null){
            nodeTb.setStatus(NodeStatus.InActive.getBoolValue());
            rootNodeService.save(nodeTb);
        }
    }
}
